package com.cn.file.ha;

import com.cn.common.ServerThread;
import com.cn.file.CommitLog;
import com.cn.file.SelectMappedBufferResult;
import com.cn.msg.Message;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;

import java.nio.ByteBuffer;

public class HAConnection {

    private HAService haService;
    private ChannelHandlerContext ctx;
    private Channel channel;
    private WriteServer writeServer;
    private CommitLog commitLog;

    public HAConnection(HAService haService, ChannelHandlerContext ctx) {
        this.haService = haService;
        this.ctx = ctx;
        this.channel = ctx.channel();
        this.writeServer = new WriteServer();
        this.commitLog = this.haService.getCommitLog();

        start();
    }

    public void start() {
        this.writeServer.start();
    }

    public void shutdown() {
    }


    //写服务
    class WriteServer extends ServerThread {

        private long nextTransferFromWhere = -1;

        @Override
        public String getServiceName() {
            return WriteServer.class.getName();
        }

        @Override
        public void run() {
            System.out.println("向slave端写入数据开始.");
            while (!isStopped()) {
                long offset = haService.getOffset();
                if (offset == -1) {
                    continue;
                }
                if (nextTransferFromWhere == -1) {
                    if (offset == 0) {
                        long commitSize = haService.getDefaultMessageStore().getMessageStoreConfig().getMappedFileSize();
                        long masterOffset = haService.getCommitOffset();
                        /**
                         * 比如：
                         *  1026-(1026%1024)=1024
                         *  commitlog文件大小是1024，那么由以上公式得出，应该在第二个commitlog文件
                         *
                         */
                        masterOffset = masterOffset - (masterOffset % commitSize);
                        if (masterOffset < 0) {
                            masterOffset = 0;
                        }
                        nextTransferFromWhere = masterOffset;
                    } else {
                        nextTransferFromWhere = offset;
                    }
                }
                if (nextTransferFromWhere < haService.getCommitOffset()) {
                    SelectMappedBufferResult result = commitLog.getData(nextTransferFromWhere);
                    Message message = new Message();
                    message.setType((byte) 21);//复制数据
                    message.setFlag((byte) 2);
                    message.setNettyRequestWrapperCode((byte) 20);
                    try {
                        byte[] bytes = new byte[result.getSize()];
                        result.getByteBuffer().get(bytes);
                        message.setByteBodys(bytes).setLength(result.getSize());
                        boolean ok = transfer(message);
                        nextTransferFromWhere = nextTransferFromWhere + result.getSize();
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        result.release();
                    }
                } else {
                    //没数据了线程等待一会儿
                    haService.getWaitNotifyObject().allWaitForRunning(200);
                }
            }
            haService.getWaitNotifyObject().remove();

            System.out.println("向slave端写入数据结束.");
        }

        public boolean transfer(Message message) {
            if (channel.isActive()) {
                channel.writeAndFlush(message).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception {
                        if (f.isSuccess()) {
                        }
                    }
                });
            }
            return false;
        }

    }

}
